package com.md.common.utils.redis;

import com.md.common.utils.redis.config.RedisProperties;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.exceptions.JedisException;

import java.util.*;

/**
 * 功能描述：redis集群操作器实现
 *
 * @author lujianping@oak168.com
 * @version 1.0
 * @date 2018-6-15 11:51
 * @since JDK 1.8
 */
@SuppressWarnings("unused")
public class RedisClusterOperator extends RedisOperator {
    private static Logger log = LoggerFactory.getLogger(RedisClusterOperator.class);

    private static BinaryJedisCluster jc = null;

    private GenericObjectPoolConfig poolConfig;

    /**
     * RedisClusterMange构造器
     */
    public static Builder builer() {
        return new Builder();
    }

    /**
     * 构建方法
     *
     * @param build 构造器
     */
    private RedisClusterOperator setBuild(Builder build) {
        this.setHosts(build.getHosts());
        this.setPassword(build.getPassword());
        this.setExpire(build.getExpire());
        this.setPort(build.getPort());
        this.setTimeout(build.getTimeout());
        this.setHost(build.getHost());
        return this;
    }

    @Override
    public void init() {
        if (jc == null) {
            Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
            String[] hosts = getHosts().split(",");
            for (String host : hosts) {
                String[] server = host.split(":");
                jedisClusterNodes.add(new HostAndPort(server[0], Integer.parseInt(server[1])));
            }
            if (poolConfig == null) {
                poolConfig = new GenericObjectPoolConfig();
            }
            jc = new BinaryJedisCluster(jedisClusterNodes, getTimeout(), getMaxRedirections(), poolConfig, getPassword());
            log.info("jedis 初始化成功!......");
        }
    }

    /**
     * redis集群配置
     * @param rp
     */
    public void setRedisConfig(RedisProperties rp) {
        this.setHosts(rp.getClusterHost());
        this.setPassword(rp.getPassword());
        poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(rp.getMaxTotal());
        poolConfig.setMaxIdle(rp.getMaxIdle());
        poolConfig.setMaxWaitMillis(rp.getMaxWaitMillis());
        poolConfig.setTestOnBorrow(rp.isTestOnBorrow());
        poolConfig.setBlockWhenExhausted(rp.isBlockWhenExhausted());
        poolConfig.setTestOnReturn(rp.isTestOnReturn());
        poolConfig.setTestWhileIdle(rp.isTestWhileIdle());
        poolConfig.setMinEvictableIdleTimeMillis(rp.getMinEvictableIdleTimeMillis());
        poolConfig.setTimeBetweenEvictionRunsMillis(rp.getTimeBetweenEvictionRunsMillis());
        poolConfig.setNumTestsPerEvictionRun(rp.getNumTestsPerEvictionRun());
    }

    /**
     * 此方法会阻塞当前线程
     *
     * @param jedisPubSub 订阅监听器对象
     * @param channels    订阅的渠道,支持多个
     */
    @Override
    public void subscribe(final JedisPubSub jedisPubSub, final String... channels) {
        jc.subscribe(jedisPubSub, channels);//阻塞  通过 jedisPubSub.unsubscribe可以取消订阅
    }

    /**
     * 此方法会阻塞当前线程
     *
     * @param jedisPubSub 订阅监听器对象
     * @param patterns    订阅的渠道模式,支持多个
     */
    @Override
    public void psubscribe(JedisPubSub jedisPubSub, String... patterns) {
        jc.psubscribe(jedisPubSub, patterns);//阻塞  通过 jedisPubSub.unsubscribe可以取消订阅
    }

    @Override
    public Long publish(final String channel, final String message) {
        return jc.publish(channel, message);
    }

    @Override
    public Long incr(String key) throws JedisException {
        return jc.incr(key);
    }

    @Override
    public Long incrBy(String key, int increment) throws JedisException {
        return jc.incrBy(key, increment);
    }

    @Override
    public Double incrByFloat(String key, double increment) throws JedisException {
        return jc.incrByFloat(key, increment);
    }

    @Override
    public Long decr(String key) throws JedisException {
        return jc.decr(key);
    }

    @Override
    public Long decrBy(String key, int decrement) throws JedisException {
        return jc.decrBy(key, decrement);
    }

    @Override
    public Map<String, String> hgetAll(String key) {
        return jc.hgetAll(key);
    }

    @Override
    public Set<String> hkeys(String key) {
        return jc.hkeys(key);
    }

    @Override
    public String hget(String key, String field) {
        return jc.hget(key, field);
    }

    @Override
    public byte[] hgetBytes(String key, String field) {
        return jc.hgetBytes(key, field);
    }

    @Override
    public long hsetBytes(String key, String field, byte[] value) {
        return jc.hset(key, field, value);
    }

    @Override
    public List<String> hmget(String key, String... fields) {
        return jc.hmget(key, fields);
    }

    @Override
    public Long hIncrBy(String key, String field, Integer value, Integer max, Integer min){
        return jc.hIncrBy(key, field, value, max, min);
    }
    
    @Override
    public Long hset(String key, String field, String value) {
        return jc.hset(key, field, value);
    }

    @Override
    public String hmset(String key, Map<String, String> kv) {
        return jc.hmset(key, kv);
    }

    @Override
    public Long hIncrBy(String key, String field, int increment) {
        return jc.hincrBy(key, field, increment);
    }

    @Override
    public boolean hsetnx(String key, String field, String value) {
        return jc.hsetnx(key, field, value) == 1;
    }

    @Override
    public boolean hexists(String key, String field) {
        return jc.hexists(key, field);
    }

    @Override
    public Long hdel(String key, String... fields) {
        return jc.hdel(key, fields);
    }

    @Override
    public Long hlen(String key) {
        return jc.hlen(key);
    }

    @Override
    public String getAndSet(String key, String value) {
        return jc.getSet(key, value);
    }

    @Override
    public Set<String> keys(String pattern) {
        return jc.keys(pattern);
    }

    @Override
    public long ttl(String key) {
        return jc.ttl(key);
    }

    @Override
    public byte[] get(String key) {
        return jc.getBytes(key);
    }

    @Override
    public void set(String key, byte[] value) {
        jc.set(key, value);
    }

    @Override
    public void set(String key, byte[] value, int expire) {
        if (expire != 0) {
            jc.setex(key, expire, value);
        } else {
            jc.set(key, value);
        }
    }

    @Override
    public boolean exists(String key) {
        if (key == null || key.trim().equals("")) {
            return false;
        }
        return jc.exists(key);
    }

    @Override
    public Long setExpire(String key, int expire) {
        return jc.expire(key, expire);
    }

    @Override
    public Long llen(String key) {
        return jc.llen(key);
    }

    @Override
    public String lset(String key, long index, String value) {
        return jc.lset(key, index, value);
    }

    @Override
    public String ltrim(String key, long start, long end) {
        return jc.ltrim(key, start, end);
    }

    @Override
    public String lindex(String key, long index) {
        return jc.lindex(key, index);
    }

    public Long lrem(String key, int count, String value) {
        if (key == null || key.trim().equals("") || value == null) {
            return 0l;
        }
        return jc.lrem(key, count, value);
    }

    public Long lpush(String key, String value) {
        if (key == null || key.trim().equals("") || value == null) {
            return 0l;
        }
        return jc.lpush(key, value);
    }

    public Long rpush(String key, String value) {
        if (key == null || key.trim().equals("") || value == null) {
            return 0l;
        }
        return jc.rpush(key, value);
    }

    public String rpop(String key) {
        if (key == null || key.trim().equals("")) {
            return "";
        }
        return jc.rpop(key);
    }

    public String lpop(String key) {
        if (key == null || key.trim().equals("")) {
            return "";
        }
        return jc.lpop(key);
    }

    public List<String> lrange(String key, long start, long stop) {
        if (key == null || key.trim().equals("") || start > stop) {
            return new ArrayList<>(0);
        }
        return jc.lrange(key, start, stop);
    }

    @Override
    public boolean setnx(String key, byte[] value, int expireSec) {
        Long setnx = jc.setnx(key, value);
        if (expireSec > 0) {
            jc.expire(key, expireSec);
        }
        return setnx != null && setnx == 1;
    }

    @Override
    public boolean setnx(String key, String value, int expireSec) {
        Long setnx = jc.setnx(key, value);
        if (expireSec > 0) {
            jc.expire(key, expireSec);
        }
        return setnx != null && setnx == 1;
    }

    @Override
    public Long del(String key) {
        return jc.del(key);
    }

    @Override
    public Long del(String... keys) {
        Long count = 0l;
        for (String key : keys) {
            Long result = this.del(key);
            count = count + result;
        }
        return count;
    }

    @Override
    public <T> T getObject(String key, Class<T> clazz) {
        if (get(key) == null) {
            return null;
        }
        return KryoSerialize.deserialize(get(key), clazz);
    }

    @Override
    public String getString(String key) {
        return jc.get(key);
    }

    @Override
    public void setString(String key, String value) {
        jc.set(key, value);
    }

    @Override
    public void setString(String key, String value, int expire) {
        if (expire != 0) {
            jc.setex(key, expire, value);
        } else {
            jc.set(key, value);
        }
    }

    @Override
    public void setObject(String key, Object object) {
        set(key, KryoSerialize.serialize(object));
    }

    @Override
    public void setObject(String key, Object object, int expire) {
        if (expire != 0) {
            jc.setex(key, expire, KryoSerialize.serialize(object));
        } else {
            set(key, KryoSerialize.serialize(object));
        }
    }

    @Override
    public List<String> blpop(int timeout, String key) {
        return jc.blpop(timeout < 0 ? 0 : timeout, key);
    }

    @Override
    public List<String> brpop(int timeout, String key) {
        return jc.brpop(timeout < 0 ? 0 : timeout, key);
    }

    @Override
    public String brpoplpush(String source, String destination, int timeout) {
        throw new UnsupportedOperationException("jedis Cluster mode not support brpoplpush");
    }

    @Override
    public String rpoplpush(String source, String destination) {
        throw new UnsupportedOperationException("jedis Cluster mode not support rpoplpush");
    }

    @Override
    public Double zscore(String key, String member) {
        return jc.zscore(key, member);
    }

    @Override
    public Set<String> zrevrange(String key, long start, long end) {
        return jc.zrevrange(key, start, end);
    }

    @Override
    public Long zadd(String key, double score, String member) {
        return jc.zadd(key, score, member);
    }

    @Override
    public Long zadd(String key, Map<String, Double> scoreMembers) {
        return jc.zadd(key, scoreMembers);
    }

    @Override
    public Set<String> zrange(String key, long start, long end) {
        return jc.zrange(key, start, end);
    }

    @Override
    public Long zrem(String key, String... members) {
        return jc.zrem(key, members);
    }

    @Override
    public Double zincrby(String key, double score, String member) {
        return jc.zincrby(key, score, member);
    }

    @Override
    public Set<String> zrangeByScore(String key, double min, double max, int offset, int count) {
        return jc.zrangeByScore(key, min, max, offset, count);
    }

    @Override
    public Set<String> zrangeByScore(String key, double min, double max) {
        return jc.zrangeByScore(key, min, max);
    }

    @Override
    public Set<String> zrevrangeByScore(String key, double max, double min) {
        return jc.zrevrangeByScore(key, max, min);
    }

    @Override
    public Set<String> zrevrangeByScore(String key, double max, double min, int offset, int count) {
        return jc.zrevrangeByScore(key, max, min, offset, count);
    }

    @Override
    public Long zremrangeByScore(String key, double start, double end) {
        return jc.zremrangeByScore(key, start, end);
    }

    @Override
    public Long lpushx(String key, String value) {
        return jc.lpushx(key, value);
    }

    @Override
    public Long rpushx(String key, String value) {
        return jc.rpushx(key, value);
    }

    @Override
    public Long zremrangeByRank(String key, long start, long end) {
        return jc.zremrangeByRank(key, start, end);
    }

    @Override
    public Long zcount(String key, double min, double max) {
        return jc.zcount(key, min, max);
    }

    public void setPoolConfig(GenericObjectPoolConfig poolConfig) {
        this.poolConfig = poolConfig;
    }

    @Override
    public void sadd(String key, String... members) {
        jc.sadd(key, members);
    }

    @Override
    public Set<String> sget(String key) {
        return jc.smembers(key);
    }

    @Override
    public void sremove(String key, String... members) {
        jc.srem(key, members);

    }

    /**
     * 构造参数
     */
    public static class Builder {
        private String host = "127.0.0.1:7000";
        private String hosts = "127.0.0.1:7000,127.0.0.1:7001";
        private int port = 7000;
        private int expire = 0;
        private int timeout = 0;
        private String password = "";

        private Builder() {
        }

        public String getHost() {
            return host;
        }

        public Builder setHost(String host) {
            this.host = host;
            return this;
        }

        public String getHosts() {
            return hosts;
        }

        /**
         * 集群设置
         *
         * @param hosts 集群主机
         * @return 构造器
         */
        public Builder setHosts(String hosts) {
            this.hosts = hosts;
            return this;
        }

        public int getPort() {
            return port;
        }

        /**
         * 单台设置
         *
         * @param port 端口
         * @return 构造器
         */
        public Builder setPort(int port) {
            this.port = port;
            return this;
        }

        public int getExpire() {
            return expire;
        }

        /**
         * 单台设置
         *
         * @param expire 过期时间
         * @return 构造器
         */
        public Builder setExpire(int expire) {
            this.expire = expire;
            return this;
        }

        public int getTimeout() {
            return timeout;
        }

        /**
         * 单台设置
         *
         * @param timeout 连接超时时间
         * @return 构造器
         */
        public Builder setTimeout(int timeout) {
            this.timeout = timeout;
            return this;
        }

        public String getPassword() {
            return password;
        }

        /**
         * 单台设置
         *
         * @param password 密码
         * @return 构造器
         */
        public Builder setPassword(String password) {
            this.password = password;
            return this;
        }

        public RedisClusterOperator build() {
            return new RedisClusterOperator().setBuild(this);
        }
    }

    @Override
    public void flushDB() {
        jc.flushDB();
    }
}
